
package com.bangcommunity.bbframe.common.utils.thread;

import com.bangcommunity.bbframe.common.utils.lang.SystemUtils;

import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author tanghc
 */
public class CommonThreadPoolManager {

    // 线程池维护线程的最少数量
    private static int CORE_SIZE = 16 * SystemUtils.getAvailableProcessors();

    // 线程池维护线程的最大数量
    private static int MAX_SIZE = 32 * SystemUtils.getAvailableProcessors();

    // 线程池维护线程所允许的空闲时间
    private static int KEEP_ALIVE_TIME = 0;

    // 线程池所使用的缓冲队列大小
    private static int WORK_QUEUE_SIZE = 100;

    // 任务调度周期
    private static int SCHEDULE_TASK_QOS_PERIOD = 10;
    // 任务调度周期
    private static int SCHEDULE_CORE_SIZE = 1;
    private static boolean daemo = true;
    /*
     * 创建一个调度线程池
     */
    static ScheduledExecutorService scheduler = null;
    /*
     * 线程池超出界线时将任务加入缓冲队列
     */
    static final RejectedExecutionHandler handler = new RejectedExecutionHandler() {

        @Override
        public void rejectedExecution(Runnable task, ThreadPoolExecutor executor) {

            taskQueue.offer(task);
        }
    };

    /*
     * 将缓冲队列中的任务重新加载到线程池
     */
    static final Runnable accessBufferThread = new Runnable() {

        @Override
        public void run() {

            if (hasMoreAcquire()) {
                threadPool.execute(taskQueue.poll());
            }
        }
    };

    /*
     * 线程池
     */
    static ThreadPoolExecutor threadPool = null;

    // 任务缓冲队列
    private static Queue<Runnable> taskQueue = new LinkedList<Runnable>();
    static {
        threadPool = new CustomeThreadPoolExecute(CORE_SIZE, MAX_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(WORK_QUEUE_SIZE), new NamedThreadFactory("BBFRAME", daemo), handler);
        scheduler = Executors.newScheduledThreadPool(SCHEDULE_CORE_SIZE);
        /*
         * 通过调度线程周期性的执行缓冲队列中任务
         */
        scheduler.scheduleAtFixedRate(accessBufferThread, 0, SCHEDULE_TASK_QOS_PERIOD, TimeUnit.MILLISECONDS);
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            @Override
            public void run() {
                doShutDown();
            }
        }));
    }

    public void addThread(Runnable task) {

        addExecuteTask(task);
    }

    public void addThreads(List<? extends Runnable> tasks) {

        for (Runnable task : tasks) {
            addExecuteTask(task);
        }
    }

    /*
     * 消息队列检查方法
     */
    private static boolean hasMoreAcquire() {

        return !taskQueue.isEmpty();
    }

    /*
     * 向线程池中添加任务方法
     */
    public void addExecuteTask(Runnable task) {

        if (task != null) {
            threadPool.execute(task);
        }
    }

    public void shutdown() {

        doShutDown();
    }

    private static void doShutDown() {

        if (null != threadPool) {
            threadPool.shutdown();
            threadPool = null;
        }
        if (null != scheduler) {
            scheduler.shutdown();
            scheduler = null;
        }
    }

    /*
     * 将构造方法访问修饰符设为私有，禁止任意实例化。
     */
    private CommonThreadPoolManager() {

    }

    private static class QtThreadPoolManagerHolder {

        private static CommonThreadPoolManager instance = new CommonThreadPoolManager();
    }

    public static CommonThreadPoolManager getInstance() {

        return QtThreadPoolManagerHolder.instance;
    }
}
